package com.yc.bigdata.flink.demo.util;

import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.api.RpcClient;
import org.apache.flume.api.RpcClientFactory;
import org.apache.flume.event.EventBuilder;

import java.nio.charset.Charset;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Timestamp;

/**
 * <p></p>
 *
 * @author: YuanChilde
 * @date: 2020-02-14 14:09
 * @version: 1.0
 * Modification History:
 * Date    Author      Version     Description
 * -----------------------------------------------------------------
 * 2020-02-14 14:09    YuanChilde     1.0        新增
 */
public class FlumeAvroSendDataFromMysql {

    static final String DB_URL = "jdbc:mysql://localhost:3306/***";  //输入DB名称
    static final String USER = "***";      //DB用户名
    static final String PASS = "***";    //DB密码

    public static void main(String[] args) {
        MyRpcClientFacade client = new MyRpcClientFacade();
        client.init("localhost", 41414);
        Connection conn = null;
        Statement stmt = null;    //真实场景使用PreparedStatement防止SQL注入
        try{
            Class.forName("com.mysql.jdbc.Driver");                 // 注册 JDBC 驱动
            conn = DriverManager.getConnection(DB_URL,USER,PASS);   // 打开链接
            client.sendDataToFlume("Connect to db");
            stmt = conn.createStatement();                          // 执行查询
            for(int i = 0;i < 10;i++){
                int index = (int)(Math.random()*10) + 1;
                String sql = "SELECT * FROM avro where id=" + index;
                ResultSet rs = stmt.executeQuery(sql);              // 保存到结果集
                while(rs.next()){
                    int id  = rs.getInt("id");
                    String name = rs.getString("name");
                    Timestamp createdt = rs.getTimestamp("createdt");
                    System.out.print("ID: " + id);
                    System.out.print(", 名称: " + name);
                    System.out.print(", 创建时间: " + createdt);
                    System.out.print("\n");
                    //client.sendDataToFlume 发送数据！
                    client.sendDataToFlume("id: " + id + ", name: " + name + ", createdt: " + createdt);
                }
                rs.close();
                try {
                    Thread.sleep(1000);     //等待一秒,模拟增量场景
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            stmt.close();
            conn.close();
        }catch(SQLException se){  // 处理 JDBC 错误
            se.printStackTrace();
        }catch(Exception e){  // 处理 Class.forName 错误
            e.printStackTrace();
        }finally{  // 关闭资源
            try {
                if (stmt != null) {
                    stmt.close();
                }
            } catch (SQLException se2) {
            }
            try {
                if (conn != null) {
                    conn.close();
                }
            } catch (SQLException se) {
                se.printStackTrace();
            }
        }
        client.sendDataToFlume("avro结束");    //测试中文是否乱码：是
        client.sendDataToFlume("avro over");
        System.out.println("avro结束");
        client.cleanUp();
    }

    static class MyRpcClientFacade {
        private RpcClient client;
        private String hostname;
        private int port;

        public void init(String hostname, int port) {
            // Setup the RPC connection
            this.hostname = hostname;
            this.port = port;
            this.client = RpcClientFactory.getDefaultInstance(hostname, port);      //创建avro客户端
            // Use the following method to create a thrift client (instead of the above line):
            // this.client = RpcClientFactory.getThriftInstance(hostname, port);    //创建Thrift客户端
        }

        public void sendDataToFlume(String data) {
            // Create a Flume Event object that encapsulates the sample data
            // 调用EventBuilder重载的withBody()方法。
            Event event = EventBuilder.withBody(data, Charset.forName("UTF-8"));
            try {
                client.append(event);     // Send the event 发送数据
            } catch (EventDeliveryException e) {
                // clean up and recreate the client 清理并重新创建客户端
                client.close();
                client = null;
                client = RpcClientFactory.getDefaultInstance(hostname, port);
                // Use the following method to create a thrift client (instead of the above line):
                // this.client = RpcClientFactory.getThriftInstance(hostname, port);
            }
        }

        public void cleanUp() {
            // Close the RPC connection
            client.close();
        }
    }
}
